package com.flying.function;

import org.springframework.messaging.Message;
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import java.time.Duration;
import java.util.function.Function;

public class FunctionLib {

    public static Function<Message<String>, String> uppercase() {
        return v -> {
            System.out.println("Uppercasing:" + v);
            return v.getPayload().toUpperCase();
        };
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    public static Function<Flux<Integer>, Tuple2<Flux<String>, Flux<String>>> router() {
        return flux -> {
            Flux<Integer> connectedFlux = flux.publish().autoConnect(2);
            UnicastProcessor even = UnicastProcessor.create();
            UnicastProcessor odd = UnicastProcessor.create();
            Flux<Integer> evenFlux = connectedFlux.filter(number -> number % 2 == 0)
                    .doOnNext(number -> even.onNext("Even:" + number));
            Flux<Integer> oddFlux = connectedFlux.filter(number -> number % 2 != 0)
                    .doOnNext(number -> odd.onNext("Odd:" + number));
            return Tuples.of(Flux.from(even).doOnSubscribe(x -> evenFlux.subscribe()),
                    Flux.from(odd).doOnSubscribe(x -> oddFlux.subscribe()));
        };
    }

    public static Function<Flux<Integer>, Flux<Integer>> aggregator() {
        return flux ->
                flux.window(Duration.ofMillis(5000))
                        .flatMap(window -> window.filter(v -> v % 2 == 0)
                                .reduce((a, b) -> a + b))
                        .doOnNext(System.out::println);
    }
}
